import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.jobcontrol.Job;

public class Projeto {

	/*
	 * Esse map recebe como entrada a base de dados e retorna para o reduce como
	 * chave o userid e como valor as fotos que o usu�rio fez upload e os seus
	 * contatos. Entrada: key: N�mero da linha no arquivo da base de dados
	 * value: Linha do arquivo da base de dados, no formato:
	 * userid,otheruserid/photoid,event,timestamp Sa�da: key: userid value:
	 * photoid,otherUserId1
	 */
	public static class Map extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {

		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
			String line = value.toString();
			String[] lineSplited = line.split(",");

			if (lineSplited.length != 4) {
				System.out.println("MAP1-ERRO: values.length != 4");
			} else if (lineSplited[0].isEmpty()) {
				System.out.println("MAP1-ERRO: userid not found: " + line);
			} else if (lineSplited[2].trim().equalsIgnoreCase("0")
					|| lineSplited[2].trim().equalsIgnoreCase("1")) {
				node.set(lineSplited[0].trim());
				word.set(lineSplited[2].trim() + "," + lineSplited[1].trim());
				if (lineSplited[1].isEmpty())
					System.out.println("MAP1-DEBUG valor de entrada: " + line);
				else
					output.collect(node, word);
			}
		}

	}

	/*
	 * Esse reduce � respons�vel por organizar fazer um merge da sa�da do map.
	 * Entrada: key: userid value: photoid1,otherUserId1 photoid2,otherUserId2
	 * ... photoidN,otherUserIdN Sa�da: key: userid value: photoid1 photoid2 ...
	 * photoidN,otherUserId1 otherUserId2 ... otherUserIdN
	 */
	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		private Text word = new Text();

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {

			List<String> photosId = new ArrayList<String>();
			List<String> otherUsersId = new ArrayList<String>();

			while (values.hasNext()) {
				String next = values.next().toString();
				String[] value = next.split(",");
				if (value.length != 2)
					System.out.println("REDUCE1-DEBUG: valor atual: " + next);

				if (value[0].trim().equalsIgnoreCase("1"))
					photosId.add(value[1]);
				else if (value[0].trim().equalsIgnoreCase("0"))
					otherUsersId.add(value[1]);
				else
					System.out.println("REDUCE1-ERRO: value[1].trim()");
			}

			String photosIdSaida = "";
			for (String str : photosId)
				photosIdSaida += str + " ";
			photosIdSaida = photosIdSaida.trim();
			if (photosIdSaida.isEmpty())
				photosIdSaida = " ";

			String otherUsersIdSaida = "";
			for (String str : otherUsersId)
				otherUsersIdSaida += str + " ";
			otherUsersIdSaida = otherUsersIdSaida.trim();
			if (otherUsersIdSaida.isEmpty())
				otherUsersIdSaida = " ";

			word.set(photosIdSaida + "," + otherUsersIdSaida);
			output.collect(key, word);
		}

	}

	/*
	 * Esse map recebe como entrada a base de dados e a sa�da do MapReduce 1 e
	 * retorna para o reduce como chave o photoid e como valor o usu�rio que fez
	 * o upload caso a entrada tenha sido do MapReduce 1, caso seja da base de
	 * dados ent�o a sa�da ter� como chave o photoid do mesmo jeito e como valor
	 * o userid do usu�rio que marcou o photoid como favorito e o timestamp.
	 * Entrada: Base de Dados: key: N�mero da linha. value: Linha, no formato:
	 * userid,otheruserid/photoid,event,timestamp Sa�da MapReduce 1: key: N�mero
	 * da linha. value: Linha, no formato: userid [tab] photoid1 photoid2 ...
	 * photoidN,otherUserId1 otherUserId2 ... otherUserIdN Sa�da: Base de Dados:
	 * key: photoid value: userId,timestamp Sa�da do MapReduce 1: key: photoid
	 * value: userId
	 */
	public static class Map2 extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {

		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
			String line = value.toString();
			String[] values = line.split(",");

			if (values.length == 2) {
				String[] userIdPhotosId = values[0].split("\t");
				String[] photosId = userIdPhotosId[1].split(" ");

				for (int i = 0; i < photosId.length; i++) {
					node.set(photosId[i].trim());
					word.set(userIdPhotosId[0].trim());
					output.collect(node, word);
				}
			} else if (values.length == 4) {
				if (values[2].equalsIgnoreCase("2")) {
					if (!values[1].isEmpty() && !values[0].isEmpty()
							&& !values[3].isEmpty()) {
						node.set(values[1].trim());
						word.set(values[0].trim() + "," + values[3].trim());
						output.collect(node, word);
					} else {
						System.out.println("MAP2-ERROR: " + line);
					}
				}
			} else {
				System.out.println("MAP2-ERRO: values.length != 2 4 - "
						+ values.length + "::" + Arrays.toString(values));
			}
		}

	}

	/*
	 * Esse reduce � respons�vel por organizar fazer um merge da sa�da do map.
	 * Entrada: key: photoid value: userid userid1,timestamp1 userid2,timestamp2
	 * ... useridN,timestampN Sa�da: key: photoid,userid value:
	 * userid1,timestamp1 userid2,timestamp2 ... useridN,timestampN
	 */
	public static class Reduce2 extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		private Text node = new Text();
		private Text word = new Text();

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {

			String photoOwner = null;
			List<String[]> marcouPhotoComoFavorito = new ArrayList<String[]>();

			while (values.hasNext()) {
				String str = values.next().toString();
				String[] valueSplit = str.split(",");

				if (valueSplit.length == 1) {
					photoOwner = valueSplit[0];
				} else if (valueSplit.length == 2) {
					marcouPhotoComoFavorito.add(valueSplit);
				} else {
					System.out.println("REDUCE2-ERRO: valueSplit.length > 2");
				}
			}

			if (photoOwner != null) {
				String saida = "";

				for (String[] strs : marcouPhotoComoFavorito)
					saida += strs[0] + "," + strs[1] + " ";
				saida = saida.trim();
				if (saida.isEmpty())
					saida += " ";

				node.set(key.toString() + "," + photoOwner);
				word.set(saida);

				output.collect(node, word);
			}
		}

	}

	/*
	 * Esse map recebe como entrada a sa�da do MapReduce 2 e emite para o reduce
	 * 3 como chave o userid e como valor o userid dos contatos que o usu�rio
	 * influ�nciou. Entrada: key: photoid,userid value: userid1,timestamp1
	 * userid2,timestamp2 ... useridN,timestampN Sa�da: key: userid value:
	 * userid1 userid2 ... useridN
	 */
	public static class Map3 extends MapReduceBase implements
			Mapper<Text, Text, Text, Text> {

		private Text node = new Text();
		private Text word = new Text();

//		private List<String> getContacts(String userid) throws IOException {
//			Configuration conf = new Configuration();
//			conf.set("fs.default.name", "hdfs://127.0.0.1:9000/");
//			FileSystem hdfs = FileSystem.get(conf);
//			InputStreamReader in = new InputStreamReader(hdfs.open(new Path(
//					"./tmp1/part-00000")));
//			BufferedReader br = new BufferedReader(in);
//
//			String strLine;
//			List<String> contactsList = new ArrayList<String>();
//
//			while ((strLine = br.readLine()) != null) {
//				String[] read = strLine.split("\t");
//				if (read[0].trim().equalsIgnoreCase(userid)) {
//					String[] dados = read[1].trim().split(",");
//					if (dados.length == 2) {
//						String[] contacts = dados[1].trim().split(" ");
//
//						for (int i = 0; i < contacts.length; i++)
//							contactsList.add(contacts[i]);
//					} else {
//						System.out
//								.println("MAP3-getContacts DEBUG: " + strLine);
//					}
//
//				}
//			}
//
//			return contactsList;
//		}

//		private boolean isContactOf(String userid, String contactid)
//				throws IOException {
//			List<String> contacts = getContacts(contactid);
//
//			return contacts.contains(userid);
//		}

		@Override
		public void map(Text key, Text value,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
			
			String[] dados = value.toString().split(",");
			if (dados.length == 2 || value.toString().endsWith(",")) {
				if(dados.length == 2) {
					String[] contacts = dados[1].split(" ");
					for (String string : contacts) {
						word.set(string);
						output.collect(key, word);
					}
				}else {
					word.set("");
					output.collect(key, word);
				}
			} else {
				String[] info = key.toString().split(",");
				String photoOwner = info[1];
				String photoID = info[0];
				node.set(photoOwner);
				word.set(photoID + "-" + value.toString());
				output.collect(node, word);
//				List<String[]> users = new ArrayList<String[]>();
//
//				String[] valueSplit = value.toString().split(" ");
//				for (int i = 0; i < valueSplit.length; i++) {
//					if (valueSplit[i].length() > 0)
//						users.add(valueSplit[i].split(","));
//				}
//
//				for (int i = 0; i < users.size(); i++) {
//					String userI = users.get(i)[0];
//
//					for (int j = 0; j < users.size(); j++) {
//						String userJ = users.get(j)[0];
//						if (!userJ.equalsIgnoreCase(userI)
//								&& Integer.valueOf(users.get(i)[1]) < Integer
//										.valueOf(users.get(j)[1])
//								&& isContactOf(userI, userJ)
//								&& !isContactOf(photoOwner, userJ)) {
//							node.set(userI);
//							word.set(userJ);
//							output.collect(node, word);
//						}
//					}
//				}
			}
		}

	}

	/*
	 * Esse reduce eh responsavel por organizar e fazer um merge contando os
	 * valores unicos da saida do map. Entrada: key: userid value: userid1
	 * userid2 ... useridN Saida: key: qtdContatosInfluenciados value: userid
	 */
	public static class Reduce3 extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter arg3)
				throws IOException {
//			List<String> influenciados = new ArrayList<String>();
//
//			while (values.hasNext()) {
//				String current = values.next().toString();
//				if (!current.isEmpty() && !influenciados.contains(current))
//					influenciados.add(current);
//			}
//
//			output.collect(new IntWritable(influenciados.size()), key);
			List<String> contacts = new ArrayList<String>();
			List<String> favoritados = new ArrayList<String>(); 
			while(values.hasNext()) {
				Text next = values.next();
				if(next.toString().contains(",")) {
					favoritados.add(next.toString());
				}else {
					contacts.add(next.toString());
				}
				output.collect(key, next);
			}
		}

	}

	public static class IntComparator extends WritableComparator {
		// Essa classe � repons�vel por alterar a ordem de compara��o, por causa
		// do (-1) ele faz a compara��o e
		// utiliza a ordem descendente

		public IntComparator() {
			super(IntWritable.class);
		}

		@Override
		public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {

			Integer v1 = ByteBuffer.wrap(b1, s1, l1).getInt();
			Integer v2 = ByteBuffer.wrap(b2, s2, l2).getInt();

			return v1.compareTo(v2) * (-1);
		}
	}

	public static void main(String[] args) throws IOException {
		JobConf conf = new JobConf(Projeto.class);

		conf.setJobName("Projeto");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapperClass(Map.class);
		conf.setReducerClass(Reduce.class);

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		FileInputFormat.setInputPaths(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path("./tmp1"));

		JobClient.runJob(conf);

		JobConf conf2 = new JobConf(Projeto.class);

		conf2.setJobName("Projeto2");
		conf2.setOutputKeyClass(Text.class);
		conf2.setOutputValueClass(Text.class);

		conf2.setMapperClass(Map2.class);
		conf2.setReducerClass(Reduce2.class);

		conf2.setInputFormat(TextInputFormat.class);
		conf2.setOutputFormat(TextOutputFormat.class);

		FileInputFormat.addInputPaths(conf2, args[0] + ",./tmp1/part-00000");
		FileOutputFormat.setOutputPath(conf2, new Path("./tmp2"));

		JobConf conf3 = new JobConf(Projeto.class);

		conf3.setJobName("Projeto3");
//		conf3.setOutputKeyClass(IntWritable.class);
//		conf3.setMapOutputKeyClass(Text.class);
		conf3.setOutputKeyClass(Text.class);
		conf3.setOutputValueClass(Text.class);

		conf3.setMapperClass(Map3.class);
		conf3.setReducerClass(Reduce3.class);

		conf3.setInputFormat(KeyValueTextInputFormat.class);
		conf3.setOutputFormat(TextOutputFormat.class);

		FileInputFormat.addInputPaths(conf3, "./tmp1,./tmp2");
		FileOutputFormat.setOutputPath(conf3, new Path(args[1]));

		Job job1 = new Job(conf);
		Job job2 = new Job(conf2);
		Job job3 = new Job(conf3);
		job2.addDependingJob(job1);
		job3.addDependingJob(job2);

		job2.getJobClient().runJob(conf2);
		job3.getJobClient().runJob(conf3);
	}
}
